package com.xiaoke.mapreduce.topn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class MyTopN {

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration(true);

        //让框架知道是windows异构平台运行
        configuration.set("mapreduce.app-submission.cross-platform", "true");

        configuration.set("mapreduce.framework.name", "local");

        // 点击job下面的参数根据example写
        Job job = Job.getInstance(configuration);

        job.setJarByClass(MyTopN.class);
        // job.setJar("D:\\code\\mayun_hadoop\\test\\hadoop\\target\\hadoop-hdfs-1.0-SNAPSHOT.jar");

        // Specify various job-specific parameters
        job.setJobName("topN-localhost");
        job.addCacheFile(new Path("/data/dist1/dist.txt").toUri());

        Path inputPath = new Path("/data/topn/input1");
        TextInputFormat.setInputPaths(job, inputPath);

        Path outputPath = new Path("/data/topn/output4");
        if (outputPath.getFileSystem(configuration).exists(outputPath))
            outputPath.getFileSystem(configuration).delete(outputPath, true);
        TextOutputFormat.setOutputPath(job, outputPath);

        // map设置
        job.setMapOutputKeyClass(TKey.class);
        job.setMapperClass(TMapper.class);
        job.setMapOutputValueClass(IntWritable.class);

        //分区
        job.setPartitionerClass(TPartitioner.class); //partitioner  按  年，月  分区  -》  分区 > 分组  按 年分区！！！！！！ //分区器潜台词：满足  相同的key获得相同的分区号就可以~！
        job.setSortComparatorClass(TSortComparator.class); //年，月，温度  且  温度倒序

        // reduce设置
        job.setGroupingComparatorClass(TGroupingComparator.class);//按年月分组
        job.setReducerClass(TReducer.class);

        job.waitForCompletion(true);

    }
}
